package com.sxyjhh.service.data_exchange;

import com.robert.kafka.kclient.core.KafkaProducer;
import com.sxyjhh.bean.AqDataPushRecode;
import com.sxyjhh.common.DBConfig;
import com.sxyjhh.create.api.DataApi;
import com.sxyjhh.dao.AqDataPushRecodeRepository;
import com.sxyjhh.dbutil.basic.DBQueryUtils;
import com.sxyjhh.dbutil.utils.PageInfo;
import com.sxyjhh.utils.FtpUtil;
import com.sxyjhh.utils.JedisUtil;
import com.sxyjhh.utils.redis.RedisOperator;
import com.sxyjhh.utils.util.Md5Util;
import com.sxyjhh.utils.util.XaUtil;
import net.sf.json.JSONObject;
import org.apache.commons.codec.binary.Base64;
import org.apache.commons.collections.map.HashedMap;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.http.client.utils.CloneUtils;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;

import java.io.IOException;
import java.sql.SQLException;
import java.util.*;

/**
 * Created by Administrator on 2018/6/22.
 */
@Service
public class ImageService {

    private FTPClient ftpClient = new FTPClient();

    @Autowired
    private JedisCluster jedisCluster;

    @Autowired
    private RedisOperator redisOperator;

    @Autowired
    private AqDataPushRecodeRepository aqDataPushRecodeRepository;

    private String dbName = DBConfig.readPop("output.datasource.db.name");

    private String ip = DBConfig.readPop("output.datasource.ip");
    private String userName = DBConfig.readPop("output.datasource.username");
    private String password = DBConfig.readPop("output.datasource.password");
    private String port = DBConfig.readPop("output.datasource.port");
    private String dbSType = DBConfig.readPop("output.datasource.type");

    private static final Logger log = Logger.getLogger(ImageService.class);

    @Autowired
    private KafkaProducer fafkaProducer;

    @Autowired
    private DBQueryUtils dbQueryUtils;

    @Autowired
    PushService pushService;

    public void getImageInfo(String tableName, String outTabName, DataApi ri, String key, int pushNum) {

        Date sTime = new Date();

        try {
            FtpUtil.getInstance().connect(ftpClient, "192.168.2.180", 2121, "yjhh", "a1b2c3d4!");
        } catch (IOException e) {
            e.printStackTrace();
        }

        try {
            List<String> list = ri.selectDataQ("select count(1) from " + tableName + " where FTYPE = 'photo'");

            int count = 0;

            if (XaUtil.isEmpty(list) || list.size() < 1 || "0".equals(list.get(0))) {
                return;
            }

            count = Integer.parseInt(list.get(0));

            int num = 3000;
            int batch = (count / num) + (count % num != 0 ? 1 : 0);

            int outCount = 0;
            List countList = dbQueryUtils.queryData("select count(1) as CUT from " + outTabName);
            if (!XaUtil.isEmpty(countList)) {
                JSONObject countJson = JSONObject.fromObject(countList.get(0));
                outCount = Integer.parseInt(countJson.get("CUT") + "");
            }

            AqDataPushRecode aqDataPushRecode = new AqDataPushRecode();

            String type = "IMAGE";

            aqDataPushRecode.setDataType(type);

            aqDataPushRecode.setNum(0);
            aqDataPushRecode.setTotalNum(0);

            aqDataPushRecodeRepository.save(aqDataPushRecode);

            if (outCount > 0) {
                int delNum = delRedisData(outTabName);
                log.info("删除缓存数据：" + delNum);
                int b = cacheData(outTabName, outCount, dbQueryUtils, key, 0);
                if (b == -1) {
                    return;
                }
            }

            int toNum = 0;
            for (int i = 0; i < batch; i++) {
                log.info("==========数据对比批次==========");
                log.info((i == 0 ? 0 : i * num) + "");

                String outName = outTabName;
                List<Map<String, Object>> outList = new ArrayList<>();
                Map<String, Map<String, Map<String, Object>>> recordMap = getTableData(tableName, num, i == 0 ? 0 : i * num, ri, key, outCount);
                dataCompare(recordMap, tableName, dbQueryUtils, outCount, key, outList, outName);
                try {
                    // conversionData(outName, outList, "1", dbQueryUtils, key, pushNum);
                    if (outList.size() > 0) {
                        toNum += outList.size();
                        log.info("==========数据对比批次条数==========");
                        log.info(outList.size());
                        conversionData(outName, outList, "1", dbQueryUtils, key, pushNum, aqDataPushRecode);
                    }
                } catch (CloneNotSupportedException e) {
                    e.printStackTrace();
                }
            }
            long endTime = new Date().getTime() - sTime.getTime();
            long finTime = (endTime % (60 * 60 * 1000)) / (60 * 1000);
            log.info("==========" + tableName + "==========");
            log.info("==========任务执行完成==========");
            log.info("==========任务耗时==========");
            log.info(finTime + "分钟");
            log.info("==========总推送量==========");
            log.info(toNum);
            AqDataPushRecode dataPushRecode = aqDataPushRecodeRepository.findOne(aqDataPushRecode.getId());
            dataPushRecode.setEndTime(new Date());
            dataPushRecode.setFinishedTime(finTime);
            aqDataPushRecodeRepository.save(dataPushRecode);

        } catch (SQLException e) {
            e.printStackTrace();
        }
    }


    //获取数据并计算字段MD5值
    public Map<String, Map<String, Map<String, Object>>> getTableData(String tableName, int num, int offsetS, DataApi ri, String key, int outCount) throws SQLException {
        Map<String, Map<String, Map<String, Object>>> recordMap = new HashMap<>();

        StringBuilder sql = new StringBuilder();
        sql.append(" select FID,FLINKID,FCONTENT,FCREATETIME from ");
        sql.append(tableName);
        sql.append(" where FTYPE = 'photo' order by  ");
        sql.append(key);
        sql.append(" limit  ");
        sql.append(num);
        sql.append(" offset ");
        sql.append(offsetS);
        List<String> dataList = ri.selectDataQ(sql.toString());

        if (!XaUtil.isEmpty(dataList) && dataList.size() > 0) {
            for (int i = 0; i < dataList.size(); i++) {
                Map<String, Map<String, Object>> recordList = new HashMap<>();
                Map<String, Object> dataMap = new HashedMap();

                String[] dataArray = dataList.get(i).split("&&");
                String srcPath = dataArray[2];
                if (StringUtils.isBlank(srcPath)) {
                    continue;
                }

                String path = dataArray[2].replace("../../upload/", "");

                String md5 = Md5Util.getMD5ofStr(path);
                dataMap.put("FCONTENT", path);
                dataMap.put("MD5", md5);

                String keyValue = dataArray[0];
                dataMap.put("FID", keyValue);
                dataMap.put("FLINKID", dataArray[1]);

                dataMap.put("FCREATETIME", dataArray[3]);

                recordList.put(md5, dataMap);
                recordMap.put(keyValue, recordList);

            }
        }
        return recordMap;
    }

    private String ftpDownload(String path) {


        String imgBase64 = "";
        try {
            byte[] imageByte = FtpUtil.getInstance().download(ftpClient, path);
            if (!XaUtil.isEmpty(imageByte)) {
                imgBase64 = Base64.encodeBase64String(imageByte);
                return imgBase64;
            } else {
                return imgBase64;
            }

        } catch (IOException e) {
            e.printStackTrace();
            return imgBase64;
        }
    }


    //数据比对判断数据状态为 新增 修改
    public void dataCompare(Map<String, Map<String, Map<String, Object>>> map, String tableName, DBQueryUtils dbQueryUtils, int outCount, String key, List<Map<String, Object>> outList, String outTabName) {

        if (outCount > 0) {
            List<Map<String, Object>> updateMd5List = new ArrayList<>();
            for (Map.Entry<String, Map<String, Map<String, Object>>> m : map.entrySet()) {
                String keyValue = outTabName + "_" + m.getKey();
                Map<String, Object> insertMap = null;
                Map<String, Map<String, Object>> subMap = m.getValue();
                String json = jedisCluster.get(keyValue);
                if (StringUtils.isBlank(json)) {
                    for (Map.Entry<String, Map<String, Object>> subM : subMap.entrySet()) {
                        insertMap = subM.getValue();
                        insertMap.put("OPERATION_FLAG", "I");
                        outList.add(insertMap);
                    }
                } else {
                    Map jsMap = JSONObject.fromObject(json);
                    String oldMD5 = jsMap.get("MD5") + "";
                    for (Map.Entry<String, Map<String, Object>> subM : subMap.entrySet()) {

                        insertMap = subM.getValue();
                        String md5 = insertMap.get("MD5") + "";
                        if (!oldMD5.equals(md5)) {

                            String imageMd5 = Md5Util.getMD5ofStr(insertMap.get("FCONTENT") + "");
                            if (!imageMd5.equals(md5)) {
                                insertMap.put("OPERATION_FLAG", "U");
                                outList.add(insertMap);
                            } else {
                                log.info("MD5计算错误的数据");
                                log.info("主键值");
                                log.info(insertMap.get("FID"));
                                Map<String, Object> md5Map = new HashMap<>();
                                md5Map.put("MD5", imageMd5);
                                md5Map.put("FID", insertMap.get("FID"));
                                updateMd5List.add(md5Map);
                            }
                        }
                        jedisCluster.del(keyValue);
                    }
                }
            }

            //修改不同的Md5值
            if (!XaUtil.isEmpty(updateMd5List)) {
                log.info("MD5计算错误的数据数量 ：" + updateMd5List.size());
                updateData(outTabName, dbQueryUtils, updateMd5List, key, "1");
            }
        } else {
            for (Map.Entry<String, Map<String, Map<String, Object>>> m : map.entrySet()) {
                Map<String, Object> insertMap = null;
                Map<String, Map<String, Object>> subMap = m.getValue();
                for (Map.Entry<String, Map<String, Object>> subM : subMap.entrySet()) {
                    insertMap = subM.getValue();
                    insertMap.put("OPERATION_FLAG", "I");
                }
                outList.add(insertMap);
            }
        }
    }

    private void conversionData(String tabName, List<Map<String, Object>> resList, String type, DBQueryUtils dbQueryUtils, String key,
                                int pushNum, AqDataPushRecode aqDataPushRecodeArg) throws CloneNotSupportedException {

        List<Map<String, Object>> insertList = null;
        List<Map<String, Object>> updateList = null;

        AqDataPushRecode aqDataPushRecode = aqDataPushRecodeRepository.findOne(aqDataPushRecodeArg.getId());
        long totalNum = aqDataPushRecode.getTotalNum();
        long num = aqDataPushRecode.getNum();
        long upNum = aqDataPushRecode.getUpdateNum();
        long inNum = aqDataPushRecode.getInsertNum();

        aqDataPushRecode.setTotalNum(totalNum + resList.size());

        resList = pushService.pushTp(tabName, resList);

        aqDataPushRecode.setNum(num + resList.size());

        for (Map<String, Object> map : resList) {

            String operationFlag = map.get("OPERATION_FLAG") + "";
            if ("I".equals(operationFlag)) {
                if (XaUtil.isEmpty(insertList)) {
                    insertList = new ArrayList<>();
                }
                Map<String, Object> resMap = cloneMap(map);

                resMap.remove("OPERATION_FLAG");
                insertList.add(resMap);
            } else if ("U".equals(operationFlag)) {
                if (XaUtil.isEmpty(updateList)) {
                    updateList = new ArrayList<>();
                }
                Map<String, Object> resMap = (Map<String, Object>) CloneUtils.clone(map);
                resMap.remove("OPERATION_FLAG");
                updateList.add(resMap);
            }
        }

        if (!XaUtil.isEmpty(insertList)) {
            aqDataPushRecode.setInsertNum(inNum + insertList.size());
            insertData(tabName, dbQueryUtils, insertList, type);
        }

        if (!XaUtil.isEmpty(updateList)) {
            aqDataPushRecode.setUpdateNum(upNum + updateList.size());
            updateData(tabName, dbQueryUtils, updateList, key, type);
        }

        aqDataPushRecodeRepository.save(aqDataPushRecode);

    }


    //新增轮询表数据
    private void insertData(String tabName, DBQueryUtils dbQueryUtils, List<Map<String, Object>> resList, String type) {

        if (XaUtil.isEmpty(resList)) {
            return;
        }

        try {
            boolean flag = dbQueryUtils.insertBatchData(tabName, dbName, resList);
            if (!flag) {
                return;
            } else {
                //向轮询表插入数据
                if ("2".equals(type)) {
                    System.out.println("数据导入成功！");
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

    //修改轮询表数据
    private void updateData(String tabName, DBQueryUtils dbQueryUtils, List<Map<String, Object>> resList, String key, String type) {

        if (XaUtil.isEmpty(resList)) {
            return;
        }

        List<String> sqls = new ArrayList<>();
        StringBuilder sql = new StringBuilder();
        for (int i = 0; i < resList.size(); i++) {
            sql.delete(0, sql.length());
            Map<String, Object> map = resList.get(i);
            String keyValue = map.get(key) + "";
            sql.append("update ");
            sql.append(tabName);
            sql.append(" set ");
            String colSql = "";
            for (Map.Entry<String, Object> m : map.entrySet()) {
                Object obj = m.getValue();
                String objType = getObjType(obj);

                if ("3".equals(dbSType)) {
                    colSql += ",`" + m.getKey() + "` =  ";
                } else {
                    colSql += "," + m.getKey() + " =  ";
                }

                if ("int".equals(objType) || "Double".equals(objType) || "Float".equals(objType) || "Long".equals(objType)) {
                    colSql += obj;

                } else {
                    if (XaUtil.isEmpty(obj)) {
                        colSql += null;
                    } else {
                        colSql += " '" + obj + "' ";
                    }
                }

            }
            sql.append(colSql.substring(1, colSql.length()));
            sql.append(" where ");
            sql.append(key);
            sql.append(" = '");
            sql.append(keyValue);
            sql.append("' ");
            sqls.add(sql.toString());
        }

        try {
            String[] sqlArray = new String[]{};
            boolean flag = dbQueryUtils.execDDLSql(sqls.toArray(sqlArray));
            if (!flag) {
                return;
            } else {
                //向轮询表插入数据
                if ("2".equals(type)) {
                    System.out.println("数据导入成功！");
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

    private String getObjType(Object obj) {

        String type = "";
        if (obj instanceof Integer) {
            type = "int";
        } else if (obj instanceof String) {
            type = "String";
        } else if (obj instanceof Double) {
            type = "Double";
        } else if (obj instanceof Float) {
            type = "Float";
        } else if (obj instanceof Long) {
            type = "Long";
        }
        return type;
    }

    public void startSynDataTP() {
        String PERSON_IMAGES = DBConfig.readPop("PERSON.IMAGES");
        String PERSON_IMAGES_OUT = DBConfig.readPop("PERSON.IMAGES.OUT");

        DataApi ri = new DataApi(DBConfig.readPop("daq.task.pconnect.url"));

        getImageInfo(PERSON_IMAGES, PERSON_IMAGES_OUT, ri, "FID", 0);
    }

    private int cacheData(String outTabName, int outCount, DBQueryUtils dbQueryUtils, String key, int b) {
        if (b == 3) {
            return -1;
        }
        int num = 5000;
        int batch = (outCount / num) + (outCount % num != 0 ? 1 : 0);
        int count = 0;
        for (int i = 0; i < batch; i++) {
            try {
//                Pipeline p = jedis.pipelined();
                PageInfo pageInfo = dbQueryUtils.queryTableDataByPaging(dbName, outTabName, i + 1, num, "select * from " + outTabName);
                List<Map> resultlist = pageInfo.getResultList();
                for (Map resMap : resultlist) {
                    String keyValue = resMap.get(key) + "";
                    JSONObject json = JSONObject.fromObject(resMap);
//                    p.set(outTabName + "_" + keyValue, json.toString());
                    jedisCluster.set(outTabName + "_" + keyValue, json.toString());
                }
                count += 1;
//                p.sync();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        Set<String> keys = redisOperator.keys(outTabName + "*");
        if (keys.size() != outCount) {
            keys.clear();
            b += 1;
            cacheData(outTabName, outCount, dbQueryUtils, key, b);
        }
        return count;
    }

    private Map<String, Object> cloneMap(Map<String, Object> resMap) {
        Map<String, Object> cloneMap = new HashMap<>();
        for (Map.Entry<String, Object> map : resMap.entrySet()) {
            Object key = map.getKey();
            Object value = map.getValue();
            cloneMap.put(key + "", value);
        }
        return cloneMap;
    }

    //删除信息
    public int delRedisData(String tabName) {
        //删除记录
        Set<String> set = redisOperator.keys(tabName + "*");
        int i = 0;
        if (!XaUtil.isEmpty(set)) {
            for (String redisKey : set) {
                i += jedisCluster.del(redisKey);
            }
        }
        return i;
    }
}
